package async_queue

import (
	"context"
	_ "embed"
	"encoding/json"
	"errors"
	"github.com/go-redis/redis/v8"
	"sync"
)

var (
	ErrQueueNotExist = errors.New("队列不存在")
	ErrMaxLength     = errors.New("超过最大长度限制")
	//ErrPushMessage   = errors.New("发送消息失败")
)

//go:embed script/push.lua
var pushScript string

//go:embed script/ack.lua
var ackScript string

type Queue struct {
	// redis客户端
	client redis.Cmdable
	// 队列，list数据结构存储
	name string
	// 队列配置及统计， hset结构存储
	configName string
	// 最大数量名称
	maxCountName string
	// 当前数量名称
	currentName string
	// 死信队列名称
	deadLetterName string
	// 防止重复关闭
	once sync.Once
	// 关闭通道
	closeCh chan struct{}
}

// Publish 尝试发布消息
// 判断队列是否存在
// 判断当前数据数量是否超过最大数据限制，HGET
// 推送数据，LPUSHX
// 当前数量计数+1 HINCRBY
func (q *Queue) Publish(ctx context.Context, data []byte) error {
	cnt, err := q.client.Exists(ctx, q.name).Uint64()
	if err != nil {
		return err
	}

	if cnt == 0 {
		// 队列不存在
		return ErrQueueNotExist
	}

	// TODO 为每一条消息生成唯一的标识ID，防止重复消费
	res, err := q.client.Eval(ctx, pushScript,
		[]string{q.name, q.configName, q.maxCountName, q.currentName}, data).
		Int64()
	if err != nil {
		return err
	}

	if res == -1 {
		// 超过最大数量
		return ErrMaxLength
	}

	return nil
}

// Consume 消费正常队列中的数据
func (q *Queue) Consume(ctx context.Context, msgCh chan<- []byte) error {
	for {
		select {
		case <-q.closeCh:
			// 通道关闭
			close(msgCh)
			return nil
		default:
			// 消费数据，原子操作
			data, err := q.client.BRPopLPush(ctx, q.name, q.deadLetterName, 0).Result()
			if errors.Is(err, context.DeadlineExceeded) {
				// 超时
				continue
			}

			if err != nil {
				return err
			}

			bs, err := json.Marshal(data)
			if err != nil {
				return err
			}

			// 推送数据
			msgCh <- bs
		}
	}
}

// ACK 消费数据后回复确认
// 删除死信队列中的数据
// 当前计数-1
func (q *Queue) ACK(ctx context.Context) error {
	_, err := q.client.Eval(ctx, ackScript, []string{q.deadLetterName, q.configName, q.currentName}).Int64()
	if err != nil {
		return err
	}
	// TODO 回复确认
	return nil
}

// DeadLetter 消费死信队列数据
func (q *Queue) DeadLetter(ctx context.Context) {

}

// Close 关闭消息队列
func (q *Queue) Close() {
	q.once.Do(func() {
		close(q.closeCh)
	})
}
